<?php
/*************************************
* PHP amqp(RabbitMQ) Demo - consumer
* Author: clx
* Date: 2015/12/3
*use method
*$a = new RabbitMQ();
*$data = array(
*	array('key'=>'msg','content'=>'测试发送短信','tel'=>'111111111'),
*	array('key'=>'msg','content'=>'测试发送短信','tel'=>'222222222'),
*	array('key'=>'msg','content'=>'测试发送短信','tel'=>'33333333')
*	
*);
*$a ->start();
*echo $a ->get($data);
*$a ->get();
*************************************/
class RabbitMQ{
	private $_config =array();
	public  $exchangeName = 'exchangeName_1'; //交换机的名字
	public  $routeName = 'routeName_1'; //路由key
	public  $queueName = 'queueName_1'; //队列名字
	public  $conn = '';//AMPQ 链接
	public  $channel = '';//管道链接
	public  $queue = '';//队列
	public  $exchange = ''; //交换机
	public function __construct($config=''){
		if(empty($config)){
			$this->config = array(
								'host' => '127.0.0.1',
								'port' => '5672',
								'login' => 'guest',
								'password' => 'guest',
								'vhost'=>'/'
							);
		}else{
			$this->config = $config;
		}		
		
	}
	/**
	*启动
	*/
	public function start(){
		
		$this->_setAMQPConnet();
		$this->_setChannel();
		$this->setExchange();
		$this->setQueue();
	}
	/*
	*设置链接
	*/
	private function _setAMQPConnet(){
		$conn = new AMQPConnection($this->config); 		
		if (!$conn->connect()) {	
			die("Cannot connect to the broker!\n");
		}else{
			$this->conn = $conn;
		}
	}
	private function _setChannel(){
		
		$this->channel = new AMQPChannel( $this->conn );
		
	}
	public function setExchange(){
		
		$this->exchange = new AMQPExchange($this->channel);
		$this->exchange ->setName($this->exchangeName);
		$this->exchange ->setType(AMQP_EX_TYPE_DIRECT); //direct类型
		$this->exchange ->setFlags(AMQP_DURABLE); //持久化
		$this->exchange ->declare();
	}
	public function setQueue(){
		//创建队列
		$this->queue = new AMQPQueue($this->channel);
		$this->queue->setName($this->queueName);
		$this->queue->setFlags(AMQP_DURABLE); //持久化
		$this->queue->declare();
		$this->queue->bind($this->exchangeName, $this->routeName);
		
	}
	public function  set($data){
			$this->channel->startTransaction();
			if(is_array($data)){
				foreach($data as $k =>$v){					
					$rst = $this->exchange->publish(json_encode($v),$this->routeName);
					if($rst == false){
						$this->channel ->rollbackTransaction();
						return false;
					}
				}
			}else{
					$rst = $this->exchange->publish(json_encode($data),$this->routeName);
				
			}
			if($rst){
				$this->channel->commitTransaction();
				return true;
			}
			
					
	}
	public function  get(){
		//消息获取
		$messages = $this->queue->get(AMQP_AUTOACK) ;
		if ($messages){
			return $messages->getBody();
		}else{
			echo '没有消息内容';
		}
	}
	
}

